import os.path
import pickle
import time
from imoocdb.catalog.entry import catalog_table, catalog_index, catalog_function
from imoocdb.common.fabric import TableColumn
from imoocdb.constant import TEMP_DIRECTORY
from imoocdb.errors import ExecutorCheckError, RollbackError
from imoocdb.session_manager import get_current_session_id
from imoocdb.sql.logical_operator import Condition, InsertOperator, UpdateOperator, DeleteOperator
from imoocdb.storage.entry import (table_tuple_get_all,
                                   table_tuple_insert_one,
                                   index_tuple_get_range,
                                   index_tuple_get_equal_value,
                                   covered_index_tuple_get_range,
                                   covered_index_tuple_get_equal_value,
                                   index_tuple_insert_one,
                                   table_tuple_get_one,
                                   table_tuple_update_one,
                                   index_tuple_update_one,
                                   index_tuple_delete_one,
                                   table_tuple_delete_multiple,
                                   index_tuple_create
                                   )
from imoocdb.sql.parser.ast import JoinType
from imoocdb.sql.parser.ast import CreateIndex, CreateTable
from imoocdb.catalog.table import CatalogTableForm
from imoocdb.catalog.index import CatalogIndexForm
from imoocdb.sql.utils import table_exists, column_exists


def is_condition_true(value: dict, condition):
    left = condition.left if not isinstance(condition.left, TableColumn) else value[condition.left]
    right = condition.right if not isinstance(condition.right, TableColumn) else value[condition.right]
    if condition.sign == '>':
        return left > right
    elif condition.sign == '<':
        return left < right
    elif condition.sign == '=':
        return left == right
    elif condition.sign == '!=':
        return left != right
    else:
        raise NotImplementedError()


def cast_tuple_pair_to_values(columns, tup):
    rv = {}
    for k, v in zip(columns, tup):
        rv[k] = v
    return rv


class PhysicalOperator:
    def __init__(self, name):
        self.name = name  # 算子名字

        # 如果我们采用CBO的内置公式进行优化
        # 那么，我们需要记住该算子的cost
        #  notice： 由于我们课程中涉及到SQL语句都很简单，RBO已经非常足够使用了
        # 就足以产生比较优化的执行计划，CBO在这里基本属于多余，因此这个公式值可能不一定能用到
        self.cost = 0

        # 该算子的子节点，即该算子的的来源
        self.children = []

        # 输出结构，用于给上层的算子提供输入
        # todo 该columns是list还是dict 我们后面要思考 要权衡
        self.columns = None

    def open(self):
        """用途：初始化该执行算子，例如提前缓存一些数据，
        提前获取数据库内部的某些状态信息，提前实现申请一些变量等"""
        raise NotImplementedError()

    def close(self):
        """用途：清理open方法创建的临时数据，关闭一些资源等"""
        raise NotImplementedError()

    def next(self):
        """用途：用于获取一行数据（tuple/record/et/al）
        数据库的执行过程是一个迭代器的 遍历过程，所以要用next()方法进行抽象
        用于表示这个迭代过程。
        例如：下面的例子，展示了一个迭代器的迭代过程
        children = [Iterator1,Iterator2]
        result = []
        for child in children:
            if not child.has_next():
                continue
            tuple = child.next()
            returns.append(process(tuple))
        return result
        """
        raise NotImplementedError()

    def add_child(self, operator):
        # 相比于外庙直接操作children列表的好处是
        # 1.可以面向抽象的接口编程，避免面向具体的实现进行编程
        # 2.避免霰弹式修改
        assert isinstance(operator, PhysicalOperator)
        self.children.append(operator)


class TableScan(PhysicalOperator):
    def __init__(self, table_name, condition: Condition = None):
        super().__init__('TableScan')
        self.table_name = table_name
        self.condition = condition
        self.columns = None

    def open(self):
        self.columns = []
        for column in catalog_table.select(lambda r: r.table_name == self.table_name)[0].columns:
            self.columns.append(TableColumn(self.table_name, column))

    def close(self):
        pass

    def next(self):
        # todo 暂时先不考虑条件检查
        for tup in table_tuple_get_all(self.table_name):
            if not self.condition:
                yield tup
            else:
                # t1(id,name)
                # 数据 (1,'xiaoming')
                # values {TableColumn(t1,id):1,TableColumn(t1,name):'xiaoming'}
                values = {k: tup[i] for i, k in enumerate(self.columns)}
                if is_condition_true(values, self.condition):
                    yield tup


class IndexScan(PhysicalOperator):
    def __init__(self, index_name, condition=None):
        super().__init__('IndexScan')
        self.index_name = index_name
        self.condition = condition
        self.condition_column = None
        self.constant = None
        self.tuple_get_equal_value = index_tuple_get_equal_value
        self.tuple_get_range = index_tuple_get_range

    def open(self):
        constants = []
        columns = []
        for node in (self.condition.left, self.condition.right):
            if isinstance(node, TableColumn):
                columns.append(node)
            else:
                constants.append(node)
        if not (constants and columns):
            raise ExecutorCheckError('bad scan condition')

        if len(columns) != 1 or len(constants) != 1:
            raise ExecutorCheckError('only supported one condition/value column.')
        self.condition_column = columns[0]
        self.constant = constants[0]
        self.fill_in_columns()

    def fill_in_columns(self):
        self.columns = []
        table_name = catalog_index.select(lambda r: r.index_name == self.index_name)[0].table_name
        for column in catalog_table.select(lambda r: r.table_name == table_name)[0].columns:
            self.columns.append(TableColumn(table_name, column))

    def close(self):
        pass

    def next(self):
        # index_tuple_get_range()
        if not self.condition:
            raise NotImplementedError()
        elif self.condition.sign == '=':
            for tup in self.tuple_get_equal_value(self.index_name, equal_value=(self.constant,)):
                yield tup
        elif self.condition.sign == '>':
            start = float('-inf')
            end = float('inf')
            if isinstance(self.condition.right, TableColumn):
                start = (self.constant,)
            else:
                end = (self.constant,)
            for tup in self.tuple_get_range(self.index_name, start=start, end=end):
                yield tup
        elif self.condition.sign == '<':
            start = float('-inf')
            end = float('inf')
            if isinstance(self.condition.left, TableColumn):
                end = (self.constant,)
            else:
                start = (self.constant,)
            for tup in self.tuple_get_range(self.index_name, start=start, end=end):
                yield tup
        else:
            raise NotImplementedError(f'not support operation {self.condition.sign} for {self.name}')


class CoveredIndexScan(IndexScan):

    def __init__(self, index_name, condition: Condition = None):
        super().__init__(index_name, condition)
        self.name = 'CoveredIndexScan'
        self.tuple_get_equal_value = covered_index_tuple_get_equal_value
        self.tuple_get_range = covered_index_tuple_get_range

    def fill_in_columns(self):
        # 采集上来的元组tuple结构，就是index的column结构
        self.columns = []
        table_name = catalog_index.select(lambda r: r.index_name == self.index_name)[0].table_name
        for column in catalog_index.select(lambda r: r.index_name == self.index_name)[0].columns:
            self.columns.append(TableColumn(table_name, column))

    def open(self):
        pass

    def close(self):
        pass

    def next(self):
        # covered_index_tuple_get_range()
        pass


class Materialize(PhysicalOperator):
    """
    物化算子
    """

    def __init__(self, name):
        super().__init__(name)
        # 物化的二维数组()
        self.tuples = []
        self.name = name

    def open(self):
        if len(self.children) != 1:
            raise ExecutorCheckError(f'{self.name} operator only supports one child operator')
        child = self.children[0]
        child.open()
        self.columns = child.columns  # 这里是一个引用 不是copy
        # self.sort_column_index = self.columns.index(self.sort_column)  # sort column 在tuple中的下标

    def close(self):
        child = self.children[0]
        child.close()

    def materialize(self):
        child = self.children[0]
        # 下面做一下元组的物化
        for tup in child.next():
            self.tuples.append(tup)


class Sort(Materialize):
    INTERNAL_SORT = 'internal sort'
    EXTERNAL_SORT = 'external sort'
    HEAP_SORT = ''  # 暂时先不实现 只是一个替换排序算法的过程

    def __init__(self, sort_column, asc=True):
        super().__init__('sort')
        self.sort_column = sort_column
        assert isinstance(self.sort_column, TableColumn)
        self.asc = asc
        self.method = self.INTERNAL_SORT

        self.sort_column_index = None  # sort column 在tuple中的下标

    def internal_sort(self):
        # sorted 不是原地排序
        # sorted(self.tuples, key=lambda t: t[self.sort_column_index], reverse=(not self.asc))  # Python内置的排序算法 用得就是排序算法
        # 下面这个排序是原地排序
        self.tuples.sort(key=lambda t: t[self.sort_column_index], reverse=(not self.asc))
        for tup in self.tuples:
            yield tup

    def open(self):
        super().open()
        self.sort_column_index = self.columns.index(self.sort_column)  # sort column 在tuple中的下标

    def external_sort(self):
        """
        外排序
        """
        max_part_size = 2
        chunks = [
            self.tuples[i:i + max_part_size] for i in range(0, len(self.tuples), max_part_size)
        ]

        # 创建临时目录
        if not os.path.exists(TEMP_DIRECTORY):
            os.mkdir(TEMP_DIRECTORY)

        # 下面我们要进行外排序，即把每个chunk分别排序，部分结果要放到磁盘缓存
        temp_files = []
        for i, chunk in enumerate(chunks):
            # 原地排序
            chunk.sort(key=lambda t: t[self.sort_column_index], reverse=(not self.asc))
            temp_file = os.path.join(TEMP_DIRECTORY, f'temp_sort_{get_current_session_id()}_{i}')
            with open(temp_file, 'wb') as f:
                for item in chunk:
                    f.write(pickle.dumps(item) + b'\n')  # 文件写完了
            temp_files.append(temp_file)

        # 接下来处理合并过程
        file_fds = [open(temp_file, 'rb') for temp_file in temp_files]

        # 此时我们来获取来自每个chunk第一个元素，然后把他们来进行排序
        first_items = []
        # 补偿机制
        file_fd_index = {}
        for i, file_fd in enumerate(file_fds):
            item = pickle.loads(file_fd.readline())  # 每次只读一行
            first_items.append(item)

            # 保证当前first_item 中每个元素都尽可能从所有chunk中获取
            # 有了该file_fd_index之后，我们就知道当前已经排好序的item是从那个fd中获取的了
            # 相当于一种反查机制 todo 再分析一下
            if item not in file_fd_index:
                file_fd_index[item] = []
            file_fd_index[item].append(i)

        # merge then
        first_items.sort(key=lambda t: t[self.sort_column_index], reverse=(not self.asc))
        while len(first_items) > 0:
            item = first_items.pop(0)  # 将第0个元素弹出来
            fd_index = file_fd_index[item].pop(0)

            # GC 对GC友好 代码写不写都可以
            if len(file_fd_index[item]) == 0:
                del file_fd_index[item]

            yield item
            next_item = file_fds[fd_index].readline()
            if not next_item:
                continue
            # 反序列化
            next_item = pickle.loads(next_item)
            first_items.append(next_item)
            if next_item not in file_fd_index:
                file_fd_index[next_item] = []
            file_fd_index[next_item].append(fd_index)

        # 收尾工作
        for file_fd in file_fds:
            file_fd.close()  # 关闭文件流
            os.unlink(file_fd.name)  # 删除文件

    def next(self):
        super().materialize()
        if self.method == self.INTERNAL_SORT:
            for tup in self.internal_sort():
                yield tup
        elif self.method == self.EXTERNAL_SORT:
            for tup in self.external_sort():
                yield tup
        else:
            raise NotImplementedError('not support')


class HashAgg(Materialize):

    def __init__(self, group_by_column, aggregate_function_name, aggregate_column):
        super().__init__('HashAgg')
        self.group_by_column = group_by_column
        self.aggregate_function_name = aggregate_function_name
        self.aggregate_column = aggregate_column

        assert isinstance(self.group_by_column, TableColumn)
        assert isinstance(self.aggregate_column, TableColumn)

        self.group_by_column_idx = None
        self.aggregate_column_idx = None

        # having子句 由于我们没有实现这个语法 所以此时不用过滤
        # 如果实现了having  self.having_condition = Condition()
        # 额外的信息

    def open(self):
        super().open()
        # todo 还有其他操作
        self.group_by_column_idx = self.columns.index(self.group_by_column)
        self.aggregate_column_idx = self.columns.index(self.aggregate_column)

        self.columns = (self.group_by_column, self.aggregate_column)

    def close(self):
        # child = self.children[0]
        # child.close()
        super().close()
        pass

    @staticmethod
    def _aggregate_function(name):
        # 此处也可以使用if else 或者Switch case
        # return {
        #     'count': len,
        #     'sum': sum,
        #     'max': max,
        #     'avg': (lambda x: sum(x) / len(x))
        # }.get(name, None)
        results = catalog_function.select(lambda r: r.function_name == name and r.agg_function)
        if len(results) != 1:
            raise ExecutorCheckError(f'not found the  agg function {name}')
        return results[0].callback

    def next(self):
        self.materialize()
        # 第一步 先做hash过程
        # key和value
        hash_table = {}
        for tup in self.tuples:

            key = tup[self.group_by_column_idx]
            value = tup[self.aggregate_column_idx]

            # 往哈希表放数据
            if key not in hash_table:
                hash_table[key] = [value]
            else:
                hash_table[key].append(value)

        # 哈希表构造完毕，对哈希表内部进行聚合
        for key, values in hash_table.items():
            aggregated_value = self._aggregate_function(self.aggregate_function_name)(values)
            yield (key, aggregated_value)


class NestedLoopJoin(PhysicalOperator):

    def __init__(self, join_type: str, left_table_name, right_table_name, join_condition):
        super().__init__('NestedLoopJoin')
        self.join_type = join_type
        self.left_table_name = left_table_name
        self.right_table_name = right_table_name
        self.join_condition = join_condition

    def open(self):
        assert len(self.children) == 2
        for child in self.children:
            child.open()
        self.columns = self.left_table.columns + self.right_table.columns

    @property
    def left_table(self):
        assert self.children[0].columns[0].table_name == self.left_table_name
        return self.children[0]

    @property
    def right_table(self):
        assert self.children[1].columns[1].table_name == self.right_table_name
        return self.children[1]

    def inner_join(self):
        for tup in self.cross_join():
            values = cast_tuple_pair_to_values(self.columns, tup)
            if is_condition_true(values, self.join_condition):
                yield tup

    def cross_join(self):
        """笛卡尔积"""
        # for循环 不考虑其他I/O代价，索引代价等，其复杂度就是O(M*N)
        for left_tuple in self.left_table.next():  # 在循环外面的是外表
            for right_table in self.right_table.next():  # 里面的是内表
                yield left_tuple + right_table

    def outer_join(self, outer_table, inner_table, exchange=False):
        if not exchange:
            padding_nulls = tuple(None for _ in range(len(inner_table.columns)))
        else:
            padding_nulls = tuple(None for _ in range(len(outer_table.columns)))

        for outer_tuple in self.left_table.next():
            matching_tuples = []
            for inner_tuple in self.right_table.next():
                if not exchange:
                    joined_tuple = outer_tuple + inner_tuple
                else:
                    joined_tuple = inner_tuple + outer_tuple
                values = cast_tuple_pair_to_values(self.columns, joined_tuple)
                if is_condition_true(values, self.join_condition):
                    matching_tuples.append(joined_tuple)
            if not matching_tuples:
                if not exchange:
                    matching_tuples.append(outer_tuple + padding_nulls)
                else:
                    matching_tuples.append(padding_nulls + outer_tuple)
            for tup in matching_tuples:
                yield tup

    def left_join(self):
        # 左连接和右连接是等价的
        # t1 left join t2 等价于 t2 left join t1
        #
        for tup in self.outer_join(outer_table=self.left_table, inner_table=self.right_table, exchange=False):
            yield tup

    def right_join(self):
        # 左连接和右连接是等价的
        # t1 left join t2 等价于 t2 left join t1
        for tup in self.outer_join(outer_table=self.right_table, inner_table=self.left_table, exchange=True):
            yield tup

    def full_join(self):
        #
        padding_nulls_left = tuple([None] * len(self.left_table.columns))
        padding_nulls_right = tuple([None] * len(self.right_table.columns))
        # 此处是一个物化过程，因为这些元组不止一次被使用，不像上面那些Join类型
        # 那些元组只使用一次，因此不需要物化
        left_tuples = []
        right_tuples = []
        for tup in self.left_table.next():
            left_tuples.append(tup)
        for tup in self.right_table.next():
            right_tuples.append(tup)
        # 下面开始Full，Join过程
        for left_tuple in left_tuples:
            matching_tuples = []
            for right_tuple in right_tuples:
                joined_tuple = left_tuple + right_tuple
                values = cast_tuple_pair_to_values(self.columns, joined_tuple)
                if is_condition_true(values, self.join_condition):
                    matching_tuples.append(joined_tuple)
            # 到此时相当于做完了内连接
            # 下面的部分是左连接部分
            if not matching_tuples:
                matching_tuples.append(left_tuple + padding_nulls_right)
            # 到此时 已经完成了左连接的实现
            for tup in matching_tuples:
                yield tup
        # 下面开始做右连接部分
        for right_tuple in right_tuples:
            not_matched = True
            for left_tuple in left_tuples:
                joined_tuple = left_tuple + right_tuple
                values = cast_tuple_pair_to_values(self.columns, joined_tuple)
                if is_condition_true(values, self.join_condition):
                    # 此时相当于去重
                    # 避免两次返回innerJoin的结果
                    not_matched = False
                    break
            if not_matched:
                yield padding_nulls_left + right_tuple

    def close(self):
        for child in self.children:
            child.close()

    def next(self):
        if self.join_type == JoinType.CROSS_JOIN:
            generator = self.cross_join()
        elif self.join_type == JoinType.INNER_JOIN:
            generator = self.inner_join()
        elif self.join_type == JoinType.LEFT_JOIN:
            generator = self.left_join()
        elif self.join_type == JoinType.RIGHT_JOIN:
            generator = self.right_join()
        elif self.join_type == JoinType.FULL_JOIN:
            generator = self.full_join()
        else:
            raise NotImplementedError(f'not supported {self.join_type}')
        for tup in generator:
            yield tup


class PhysicalQuery(PhysicalOperator):
    def __init__(self):
        """该物理算子相当于一个dummy，相当于一种编程技巧
        但是我们给他找点活儿干，让他帮忙记录一下执行阶段的信息，如执行耗时等
        """
        super().__init__('Result')
        self.open_time = 0
        self.close_time = 0
        self.actual_rows = 0  # 影响了多少行
        self.projection_column_ids = []

    def open(self):
        # 获取单调时间
        self.open_time = time.monotonic()
        for child in self.children:
            child.open()
        # 遍历要输出的列，寻找子节点中对应的下标位置
        child_columns = self.children[0].columns  # 子节点返回的所有列信息
        for target_column in self.columns:
            for j, child_column in enumerate(child_columns):
                if target_column == child_column:
                    self.projection_column_ids.append(j)

    def close(self):
        for child in self.children:
            child.close()
        self.close_time = time.monotonic()

    def next(self):
        for child in self.children:
            self.actual_rows += 1
            for tup in child.next():
                # 做列裁剪（投影操作）
                yield tuple(tup[i] for i in self.projection_column_ids)

    @property
    def elapsed_time(self):
        # 执行的总耗时
        return self.close_time - self.open_time


class PhysicalInsert(PhysicalOperator):
    def __init__(self, logical_operator: InsertOperator):
        super().__init__('Insert')
        self.logical_operator = logical_operator
        self.column_ids = []
        self.table_column_num = 0
        self.indexes = None

    def open(self):
        all_columns = catalog_table.select(
            lambda r: r.table_name == self.logical_operator.table_name
        )[0].columns
        for i, column in enumerate(all_columns):
            for table_column in self.logical_operator.columns:
                if column == table_column.column_name:
                    self.column_ids.append(i)
        self.table_column_num = len(all_columns)
        if len(self.column_ids) != len(self.logical_operator.columns) or (
                len(self.column_ids) > self.table_column_num
        ):
            raise ExecutorCheckError(f'error caused by columns.')

        # 遍历所有涉及到的索引,后面也要同步更新他
        indexes = catalog_index.select(
            lambda r: r.table_name == self.logical_operator.table_name
        )
        self.indexes = []
        for index_form in indexes:
            index_name = index_form.index_name
            column_ids = []
            for column_name in index_form.columns:
                column_ids.append(all_columns.index(column_name))
            self.indexes.append(
                dict(index_name=index_name, column_ids=column_ids)
            )

        # xid = transaction_mgr.session_xid()
        # table_name = self.logical_operator.table_name
        # lock_manager.acquire_lock(('table', table_name), xid, 'x')
        # for index_item in self.indexes:
        #     lock_manager.acquire_lock(('index', index_item['index_name']), xid, 'x')

    def close(self):
        # xid = transaction_mgr.session_xid()
        # table_name = self.logical_operator.table_name
        # lock_manager.release_lock(('table', table_name), xid)
        # for index_item in self.indexes:
        #     lock_manager.release_lock(('index', index_item['index_name']), xid)
        pass

    @staticmethod
    def _pad_null(tup, set_ids, total_length):
        full_tuple = [None] * total_length
        for i, value in zip(set_ids, tup):
            full_tuple[i] = value
        return tuple(full_tuple)

    def next(self):
        for tup in self.logical_operator.values:
            # 更新基础数据表
            try:
                location = table_tuple_insert_one(
                    self.logical_operator.table_name, self._pad_null(
                        tup, self.column_ids, self.table_column_num
                    )
                )
            except Exception as e:
                # 出现问题了，要回滚！
                raise RollbackError(
                    f'cannot insert data into the table {self.logical_operator.table_name}.'
                )
            # 同步更新所有涉及到的索引
            # insert into t1 (id) values ...; 补充null的场景
            # insert into t1 values ...;
            for index_info in self.indexes:
                index_tuple_insert_one(
                    index_info['index_name'],
                    key=self._pad_null(
                        tup, index_info['column_ids'], len(index_info['column_ids'])),
                    value=location)

            yield


class PhysicalUpdate(PhysicalOperator):
    def __init__(self, logical_operator: UpdateOperator):
        super().__init__('Update')
        self.logical_operator = logical_operator
        self.column_ids = []
        self.table_column_num = 0
        self.indexes = None

    def open(self):
        all_columns = catalog_table.select(
            lambda r: r.table_name == self.logical_operator.table_name
        )[0].columns
        for i, column in enumerate(all_columns):
            for table_column in self.logical_operator.columns:
                if column == table_column.column_name:
                    self.column_ids.append(i)
        self.table_column_num = len(all_columns)
        if len(self.column_ids) != len(self.logical_operator.columns) or (
                len(self.column_ids) > self.table_column_num
        ):
            raise ExecutorCheckError(f'error caused by columns.')

        # 遍历所有涉及到的索引,后面也要同步更新他
        indexes = catalog_index.select(
            lambda r: r.table_name == self.logical_operator.table_name
        )
        self.indexes = []
        for index_form in indexes:
            index_name = index_form.index_name
            column_ids = []
            for column_name in index_form.columns:
                column_ids.append(all_columns.index(column_name))
            self.indexes.append(
                dict(index_name=index_name, column_ids=column_ids)
            )

        # Update 是有子节点的，他的子节点就是Scan算子，不同于select
        # 语句，这个Scan算子返回的是位置 location, 可以用于更新
        assert len(self.children) == 1
        assert isinstance(self.children[0], LocationScan)
        for child in self.children:
            child.open()

        # xid = transaction_mgr.session_xid()
        # table_name = self.logical_operator.table_name
        # lock_manager.acquire_lock(('table', table_name), xid, 'x')
        # for index_item in self.indexes:
        #     lock_manager.acquire_lock(('index', index_item['index_name']), xid, 'x')

    def close(self):
        for child in self.children:
            child.close()

        # xid = transaction_mgr.session_xid()
        # table_name = self.logical_operator.table_name
        # lock_manager.release_lock(('table', table_name), xid, )
        # for index_item in self.indexes:
        #     lock_manager.release_lock(('index', index_item['index_name']), xid, )

    def _update_from_old_tuple(self, old_tuple):
        new_tuple = list(old_tuple)
        for i, value in zip(self.column_ids, self.logical_operator.values):
            new_tuple[i] = value
        assert len(new_tuple) == self.table_column_num
        return tuple(new_tuple)

    def next(self):
        # 该节点只有一个子节点，该写法等效于
        # `for location in self.children[0].next_location()`
        for child in self.children:
            for location in child.next():
                if location is None:
                    # 出现问题了，要回滚！
                    raise RollbackError(
                        f'cannot update data for the table {self.logical_operator.table_name}.'
                    )

                old_tuple = table_tuple_get_one(self.logical_operator.table_name,
                                                location)
                new_tuple = self._update_from_old_tuple(old_tuple)
                new_location = table_tuple_update_one(
                    self.logical_operator.table_name,
                    location,
                    new_tuple
                )
                # 同步更新所有涉及到的索引
                for index_info in self.indexes:
                    key = tuple(new_tuple[i] for i in index_info['column_ids'])
                    index_tuple_update_one(
                        index_info['index_name'],
                        key=key,
                        old_value=location,
                        value=new_location
                    )

                yield


class PhysicalDelete(PhysicalOperator):
    def __init__(self, logical_operator: DeleteOperator):
        super().__init__('Insert')
        self.logical_operator = logical_operator
        self.indexes = None

    def open(self):
        all_columns = catalog_table.select(
            lambda r: r.table_name == self.logical_operator.table_name
        )[0].columns

        # 遍历所有涉及到的索引,后面也要同步更新他
        indexes = catalog_index.select(
            lambda r: r.table_name == self.logical_operator.table_name
        )
        self.indexes = []
        for index_form in indexes:
            index_name = index_form.index_name
            column_ids = []
            for column_name in index_form.columns:
                column_ids.append(all_columns.index(column_name))
            self.indexes.append(
                dict(index_name=index_name, column_ids=column_ids)
            )

        # Delete 是有子节点的，他的子节点就是Scan算子，不同于select
        # 语句，这个Scan算子返回的是位置 location, 可以用于更新
        assert len(self.children) == 1
        assert isinstance(self.children[0], LocationScan)
        for child in self.children:
            child.open()

        # xid = transaction_mgr.session_xid()
        # table_name = self.logical_operator.table_name
        # lock_manager.acquire_lock(('table', table_name), xid, 'x')
        # for index_item in self.indexes:
        #     lock_manager.acquire_lock(('index', index_item['index_name']), xid, 'x')

    def close(self):
        for child in self.children:
            child.close()

        # xid = transaction_mgr.session_xid()
        # table_name = self.logical_operator.table_name
        # lock_manager.release_lock(('table', table_name), xid)
        # for index_item in self.indexes:
        #     lock_manager.release_lock(('index', index_item['index_name']), xid)

    def next(self):
        locations = []
        for child in self.children:
            for location in child.next():
                old_tuple = table_tuple_get_one(self.logical_operator.table_name,
                                                location)
                yield
                locations.append(location)
                for index_info in self.indexes:
                    index_name = index_info['index_name']
                    key = tuple(old_tuple[i] for i in index_info['column_ids'])
                    index_tuple_delete_one(index_name, key=key, location=location)

        table_tuple_delete_multiple(self.logical_operator.table_name,
                                    locations)


class PhysicalDDL(PhysicalOperator):
    def __init__(self, logical_operator):
        super().__init__('DDL')
        self.ast = logical_operator.ast

    @staticmethod
    def cast_to_type(type_name):
        if type_name == 'int' or type_name == 'integer':
            return int
        elif type_name == 'text':
            return str
        else:
            raise NotImplementedError(f'not supported this type {type_name}.')

    def open(self):
        if isinstance(self.ast, CreateTable):
            columns = []
            types = []
            for column, type_ in self.ast.columns:
                columns.append(column)
                types.append(self.cast_to_type(type_))
            catalog_table.insert(CatalogTableForm(
                self.ast.table.parts, columns, types))
        elif isinstance(self.ast, CreateIndex):
            index_name = self.ast.index.parts
            table_name = self.ast.table.parts
            if not table_exists(table_name):
                raise ExecutorCheckError(f'not found the table {table_name}.')

            columns = []
            for column in self.ast.columns:
                column = column.parts
                columns.append(column)
                if not column_exists(table_name, column):
                    raise ExecutorCheckError(
                        f'not found the column {column} in table {table_name}.')

            catalog_index.insert(CatalogIndexForm(index_name,
                                                  columns,
                                                  table_name))
            try:
                index_tuple_create(index_name, table_name, columns)
            except Exception as e:
                # 相当于一个小回滚
                catalog_index.delete(lambda r: r.index_name == index_name)
                raise RollbackError(e)
        else:
            raise NotImplementedError(f'not supported this type {type(self.ast)}.')

    def close(self):
        pass

    def next(self):
        yield


class CommandOperator(PhysicalOperator):
    def __init__(self, command, args=None):
        assert isinstance(command, str)
        super().__init__('Command')
        self.command = command
        if args is None:
            self.args = None
        else:
            # args 是 expr 列表，而expr是identifier，需要通过
            # parts 来获取具体的字符串内容
            self.args = [identifier.parts for identifier in args]

    def open(self):
        if self.command == 'SHOW':
            if self.args[0] == 'variables':
                self.columns = ['name', 'value']

    def close(self):
        pass

    def next(self):
        if self.command == 'CHECKPOINT':
            # checkpoint()
            yield
        elif self.command == 'SHOW':
            if self.args[0] == 'variables':
                pass
        # rows = [
        #     ('transaction_count', instr.transaction_count),
        #     ('current_xid', transaction_mgr.current_xid),
        #     ('activity_count', len(transaction_mgr.undo_mgr.active_transactions))
        # ]
        # for r in rows:
        #     yield r
        else:
            raise NotImplementedError(f'not supported this command {self.command}.')


class LocationScan(PhysicalOperator):
    def __init__(self, scan):
        super().__init__('LocationScan')
        # 该类是个代理类，真正执行扫描动作的是下面的 real_scan 算子
        # real_scan 算子可以是 IndexScan/TableScan, 但是调用的不是它们的
        # next() 方法，而是 next_location() 方法。
        assert isinstance(scan, IndexScan) or isinstance(scan, TableScan)
        self.real_scan = scan

    def open(self):
        self.columns = []  # nothing to return
        self.real_scan.open()

    def close(self):
        self.real_scan.close()

    def next(self):
        locations = []
        for location in self.real_scan.next_location():
            locations.append(location)
        # 这里缓存一下，是为了避免被删除元素之后，由于 next_location() 是动态
        # 获取下标的，tuple 下标会发生位置变换（例如 -1），
        # 导致后续扫描失败
        for location in locations:
            yield location
